package com.act.task;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import com.act.util.CommonUtils;
import com.act.util.JdbcUtil;
import com.act.util.SystemConfig;
import com.act.util.URLUtil;

public class DomainHdfsTask {
    
	private static final Log log = LogFactory.getLog(DomainHdfsTask.class);
	
	@SuppressWarnings("serial")
	public static void main( String[] args ) {
		try{
			long t1 = System.currentTimeMillis();
			String dayStr = "";
			if(args.length > 0){
				//时间格式:20171026
				dayStr = args[0];
			}
			log.info("传入的时间："+dayStr);
			
			SparkConf conf = new SparkConf();
	    	conf.setAppName("DomainHdfsTask");
	    	
	    	JavaSparkContext sc = new JavaSparkContext(conf);
	    	SQLContext sqlContext = new SQLContext(sc);
	    	
	        String schemaString = "type,isfraud,fraudtype,subfraudtype,country,province,area,host,url,urlparam,ip,state,control,isforeign,isweb,operator,md5,appname,sitetype,sitetypename,checkmonth,checkdate,checkhour,checktime,time,logtime,uuid,modifytime";  
	        List<StructField> fields = new ArrayList<StructField>();  
	        for (String fieldName : schemaString.split(",")) {  
	            StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);  
	            fields.add(field);  
	        }
	        
	        StructType schema = DataTypes.createStructType(fields); 
	        
	        String hdfsPath = SystemConfig.getProperty("fraud.hdfs.path"); 
	        
	    	JavaRDD<String> rdd = sc.textFile(hdfsPath+"/"+CommonUtils.getExecDate(dayStr)+"/*");
	    	JavaRDD<Row> rowRDD = rdd.map(new Function<String, Row>() {  
	            @Override
	            public Row call(String record){  
	                try{
	                	String[] field = record.split("\001");
		                return RowFactory.create(field[0].trim(),field[1].trim(),field[2].trim(),field[3].trim(),field[4].trim(),field[5].trim(),field[6].trim(),
		                						 field[7].trim(),field[8].trim(),field[9].trim(),field[10].trim(),field[11].trim(),field[12].trim(),field[13].trim(),
		                						 field[14].trim(),field[15].trim(),field[16].trim(),field[17].trim(),field[18].trim(),field[19].trim(),field[20].trim(),
		                						 field[21].trim(),field[22].trim(),field[23].trim(),field[24].trim(),field[25].trim(),field[26].trim(),field[27].trim());  
	                }catch(Exception e){
	                	log.error("解析当前行数据有问题，请检查对应行数数据");
	                	log.error(record);
	                	if(log.isErrorEnabled()){
	                		log.error(e.getMessage(),e);
	                	}
	                	return RowFactory.create("","","","","","","","","","","","","","","","","","","","","","","","","","","","");
	                }
	            }  
	        });
	    	
	    	DataFrame dataFrame = sqlContext.createDataFrame(rowRDD, schema);
	    	dataFrame.registerTempTable("fraud_temp");
	    	DataFrame reslutFrame = sqlContext.sql("select * from fraud_temp");

	    	JavaRDD<Map<String,Object>> domainRDD = reslutFrame.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>,Map<String,Object>>() {

				@Override
				public Iterable<Map<String,Object>> call(Iterator<Row> it){
					List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
					String host = "";
					while(it.hasNext()){
						try{
							Row row = it.next();
							host = row.getAs("host");
							String topDomain = URLUtil.getDomainName("http://" + host);
							if(StringUtils.isNotBlank(topDomain)){
								Map<String,Object> map = new HashMap<String,Object>();
								map.put("id", 0);
								map.put("type", row.getAs("type"));
								map.put("isfraud", row.getAs("isfraud"));
								map.put("fraudtype", row.getAs("fraudtype"));
								map.put("subfraudtype", row.getAs("subfraudtype"));
								map.put("country", row.getAs("country"));
								map.put("province", row.getAs("province"));
								map.put("area", row.getAs("area"));
								map.put("host",host);
								map.put("url", row.getAs("url"));
								map.put("urlparam", row.getAs("urlparam"));
								map.put("ip", row.getAs("ip"));
								map.put("state", row.getAs("state"));
								map.put("control", row.getAs("control"));
								map.put("isforeign", row.getAs("isforeign"));
								map.put("isweb", row.getAs("isweb"));
								map.put("operator", row.getAs("operator"));
								map.put("md5", row.getAs("md5"));
								map.put("appname", row.getAs("appname"));
								map.put("sitetype", row.getAs("sitetype"));
								map.put("sitetypename", row.getAs("sitetypename"));
								map.put("checkmonth", row.getAs("checkmonth"));
								map.put("checkdate", row.getAs("checkdate"));
								map.put("checkhour", row.getAs("checkhour"));
								map.put("checktime", row.getAs("checktime"));
								map.put("time", row.getAs("time"));
								map.put("logtime", row.getAs("logtime"));
								map.put("uuid",row.getAs("uuid"));
								map.put("modifytime", row.getAs("modifytime"));
								map.put("checkdate", row.getAs("checkdate"));
								map.put("isdail", "");
								map.put("isdomain","");
								map.put("location", "");
								map.put("topdomain",topDomain);
								list.add(map);
							}
						}catch(Exception e){
							log.info("当前解析出错的域名为："+host);
							if(log.isErrorEnabled()){
								log.error(e.getMessage(),e);
							}
						}
					}
					return list;
				}
			});
	    	
			log.info("===========开始数据插入=======");
			insertData(domainRDD);
			log.info("===========数据插入完成=======");
			
			//各专题汇总
			StatisticsTask.statics();
			
			sc.stop();
			
			long t2 = System.currentTimeMillis();
			log.info("=====总耗时："+((t2-t1)/1000/60)+"分钟======");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	@SuppressWarnings("serial")
	private static void insertData(JavaRDD<Map<String,Object>> rdd) {
		
		String columns = "id,type,country,province,area,host,url,urlparam,ip,state,control,isforeign,isweb,operator,md5,appname,sitetype,sitetypename,checkmonth,checkdate,checkhour,checktime,time,logtime,uuid,modifytime,isdail,isdomain,location,isfraud,fraudtype,subfraudtype,topdomain";
		String[] columnsSplit = columns.split(",");
		
		if(!rdd.isEmpty()){
			String sql = "truncate table ods_idc_t_detector_check_fraud";
			//String sql = "truncate table t_detector_check_fraud_temp";
			boolean flag = execSql(sql);
			if(flag){
				rdd.foreachPartition(new VoidFunction<Iterator<Map<String,Object>>>() {

					@Override
					public void call(Iterator<Map<String,Object>> it) {
						Connection conn = null;
						PreparedStatement ps = null;
						try {
							log.info("====当前分区插入数据开始====");
							conn = JdbcUtil.getConnection();
							
							String sql = "insert into ods_idc_t_detector_check_fraud("+columns+") values(CHECK_FRAUD_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
							//String sql = "insert into t_detector_check_fraud_temp("+columns+") values(CHECK_FRAUD_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
							conn.setAutoCommit(false);
							ps = conn.prepareStatement(sql);
							int n = 0;
							while (it.hasNext()) {
								Map<String,Object> map = it.next();
								for (int i = 0; i < map.size()-1; i++) {
									ps.setObject(i+1, map.get(columnsSplit[i+1]));
								}
								ps.addBatch();
								if (++n % 10000 == 0) {
									log.info("当前分区插入的数据量："+n);
									ps.executeBatch();
									conn.commit();
								}
							}
							ps.executeBatch();
							conn.commit();
							log.info("当前分区插入的数据总量："+n);
							log.info("====当前分区插入数据结束====");
						} catch (Exception e) {
							if (log.isErrorEnabled()) {
								log.error(e.getMessage(), e);
							}
						} finally {
							JdbcUtil.close(ps, conn);
						}
					}
				});
			}
		}
	}
	
	public static void createSeqIfNotExists(Connection conn){
		try{
			String sequenceName = "CHECK_FRAUD_SE";
			CommonUtils.getSequenceName(sequenceName,conn);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	public static boolean execSql(String sql){
    	boolean flag = true;
    	Connection conn = null;
    	PreparedStatement ps = null;
    	try{
    		conn = JdbcUtil.getConnection();
    		createSeqIfNotExists(conn);
    		conn.setAutoCommit(false);
    		ps = conn.prepareStatement(sql);
          	ps.executeQuery();
            conn.commit();
          	flag = true;
          }
          catch(SQLException e){
         	 flag = false;
         	 log.error("jdbc batch commit sql:"+sql+" error" , e);
              try {
                  conn.rollback();
              } catch (SQLException e1) {
                  log.error("jdbc rollback error" , e1);
              }
          }finally{
        	  JdbcUtil.close(ps,conn);
          }
    	return flag ;
    }
}

